package com.xinlianshiye.im.utils;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.xinlianshiye.clouds.common.exception.WebMessageException;
import com.xinlianshiye.im.ResponseBody;
import com.xinlianshiye.im.Result;
import com.xinlianshiye.im.bean.Record;
import com.xinlianshiye.im.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.websocket.Session;
import java.util.List;

@Slf4j
@Service
@Transactional
public class RedisReceiver {

    @Autowired
    private UserService userService;
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private SessionManager sessionManager;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    public void sendError(Session session, WebMessageException messageException) {
        sendError(session, messageException.getCode(), messageException.getMessage());
    }

    public void sendError(Session session, Integer code, String reason) {
        send(session, ResponseBody.initError(code, reason), 3000);
    }

    public void send(Record record) {
        Integer receiver = record.getOwnerId();
        Integer sender = record.isSender() ? record.getOwnerId() : record.getOtherId();
        List<Session> sessionList = sessionManager.getSessionByUserId(receiver);
        ResponseBody responseBody = ResponseBody.initChat(sender, receiver, record);
        for (Session session : sessionList) {
            send(session, responseBody);
        }
    }

    /**
     * 接收redis消息的方法
     */
    public void receiveMessage(String message) {
        stringRedisTemplate.convertAndSend("", message);
        log.info("收到redis消息: {}", message);
        try {
            ResponseBody responseBody = objectMapper.readValue(message, ResponseBody.class);
            sendToUser(responseBody.getReceiver(), responseBody);
        } catch (Exception e) {
            log.error("MessageBody 反序列化失败", e);
        }
    }

    public void sendToUser(List<Integer> receivers, Result result, String url) {
        receivers.forEach(receiver -> send(sessionManager.getSessionByUserId(receiver), result, url));
    }

    public void sendToUser(Integer receiver, Result result, String url) {
        send(sessionManager.getSessionByUserId(receiver), result, url);
    }

    public void send(List<Session> sessions, Result result, String url) {
        sessions.forEach(session -> send(session, result, url));
    }

    public void send(Session session, Result result, String url) {
        send(session, ResponseBody.initSuccess(url, result));
    }

    public void sendToUser(List<Integer> receivers, ResponseBody responseBody) {
        receivers.forEach(receiver -> send(sessionManager.getSessionByUserId(receiver), responseBody));
    }

    public void sendToUser(Integer receiver, ResponseBody responseBody) {
        send(sessionManager.getSessionByUserId(receiver), responseBody);
    }

    public void send(List<Session> sessions, ResponseBody responseBody) {
        sessions.forEach(session -> send(session, responseBody));
    }

    public void send(Session session, ResponseBody responseBody) {
        send(session, responseBody, 1000);
    }

    private void send(Session session, ResponseBody responseBody, Integer timeout) {
        try {
            responseBody.setReceiver(userService.getUserId(session));
            send(session, objectMapper.writeValueAsString(responseBody), timeout);
        } catch (JsonProcessingException e) {
            log.error("messageBody转字符串出错！", e);
        }
    }

    private void sendToUser(List<Integer> receivers, String text) {
        receivers.forEach(receiver -> send(sessionManager.getSessionByUserId(receiver), text));
    }

    public void sendToUser(Integer receiver, String text) {
        send(sessionManager.getSessionByUserId(receiver), text);
    }

    private void send(List<Session> sessions, String text) {
        sessions.forEach(session -> send(session, text));
    }

    private void send(Session session, String text) {
        send(session, text, 1000);
    }

    private void send(Session session, String text, Integer timeout) {
        try {
            if (!session.isOpen()) {
                sessionManager.onClose(session);
                return;
            }
            if (timeout < 0) { // timeout后放弃本次发送
                return;
            }
            if (sessionManager.getIdle().remove(session)) { // 判断session是否空闲，抢占式
                log.info("发送消息：用户[{}]: {}", userService.getUserId(session), text);
                sessionManager.getBusy().put(session, session.getAsyncRemote().sendText(text));
            } else {
                // 若session当前不在idle集合，则去busy集合中查看session上次是否已经发送完毕，即采用惰性判断
                synchronized (sessionManager.getBusy()) {
                    if (sessionManager.getBusy().containsKey(session) && sessionManager.getBusy().get(session).isDone()) {
                        sessionManager.getBusy().remove(session);
                        sessionManager.getIdle().add(session);
                    }
                }
                // 重试
                Thread.sleep(100);
                send(session, text, timeout - 100);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
